import geo.Geohash;
import geo.SerializeUtil;
import geo.po.Store;
import procon.Consumer;
import procon.Producer;
import pubsub.PrintListener;
import pubsub.PubClient;
import pubsub.SubClient;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
import redis.clients.jedis.JedisPubSub;

import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

/**
 * Created by li on 2016/4/16.
 */
public class Test {
    private JedisPool jedisPool;

    public static void main(String[] args) throws Exception {
        Test test = new Test().init();
        //发布者订阅模式
//        test.pubsub();
        //消费生产模式
//        test.procon();
        //位置搜索
        test.geo();
    }

    public Test init() {
        JedisPoolConfig config = new JedisPoolConfig();
        config.setMaxIdle(300);
        config.setTestOnBorrow(true);
        this.jedisPool = new JedisPool(config, "127.0.0.1", 6379, 10000, "123456");
        return this;
    }

    public void clearDB() {
        Jedis jedis = jedisPool.getResource();
        jedis.flushDB();
        returnJedis(jedis);
    }

    public void geo() {
        Jedis jedis = jedisPool.getResource();
        //精度（5位误差10平方千米，6位误差0.34平方千米）
        int precision = 6;
        //生成位置测试数据
        double[] qingdao = {36.1052, 120.3844};
        System.out.println("生成测试数据...");
        for (int i = 0; i < 5000; i++) {
            initLocations(jedis, qingdao[0], qingdao[1], i, precision);
            qingdao[0] += 0.001;
        }
        for (int i = 0; i < 5000; i++) {
            initLocations(jedis, qingdao[0], qingdao[1], i, precision);
            qingdao[1] += 0.001;
        }
        //搜索500米内的商家
        double[] mylocation = {36.2020, 120.3800};
        double range = 500.0;
        search(jedis, mylocation[0], mylocation[1], precision, range);
        returnJedis(jedis);
    }

    public void search(Jedis jedis, double lat, double lon, int precision, double range) {
        long start = System.currentTimeMillis();
        String geohash = Geohash.encode(lat, lon);
        String key = geohash.substring(0, precision);
        Set<Store> stores = new HashSet<Store>();
        //搜索中心区域及周围八个区域
        String[] expand = Geohash.expand(key);
        for (String s : expand) {
            Set<byte[]> bytes = jedis.smembers(s.getBytes());
            for (byte[] aByte : bytes) {
                Store store = (Store) SerializeUtil.unserialize(aByte);
                if (store != null)
                    stores.add(store);
            }

        }
        Store[] storesArray = stores.toArray(new Store[stores.size()]);
        for (int i = 0; i < storesArray.length - 1; i++) {
            for (int j = 0; j < storesArray.length - 1 - i; j++) {
                double[] latlong = storesArray[j].getLatlong();
                double distance1 = Geohash.getDistance(lat, lon, latlong[0], latlong[1]);
                storesArray[j].setDistance(distance1);
                double[] latlong2 = storesArray[j + 1].getLatlong();
                double distance2 = Geohash.getDistance(lat, lon, latlong2[0], latlong2[1]);
                storesArray[j + 1].setDistance(distance2);
                if (distance1 > distance2) {
                    Store temp = storesArray[j];
                    storesArray[j] = storesArray[j + 1];
                    storesArray[j + 1] = temp;
                }
            }
        }

        for (Store store : storesArray) {
            if (store.getDistance() > range) break;
            System.out.println("附近商家：" + store);
        }
        System.out.println("搜索耗时:" + (System.currentTimeMillis() - start) + "ms");
    }

    public void initLocations(Jedis jedis, double lat, double lon, int i, int precision) {
        //经纬度geohash处理
        String geohash = Geohash.encode(lat, lon);
        //截取精度（5位误差10平方千米，6位误差0.34平方千米）
        String key = geohash.substring(0, precision);
        Store store = new Store();
        store.setName("商家");
        store.setDescription(key + "区域");
        store.setGeohash(geohash);
        store.setLatlong(new double[]{lat, lon});
        jedis.sadd(key.getBytes(), SerializeUtil.serialize(store));
    }


    //发布者订阅者模式
    public void pubsub() throws Exception {
        final String channel = "test";
        Jedis jedisOfPub = jedisPool.getResource();
        PubClient pubClient = new PubClient(jedisOfPub);
        pubClient.pub(channel, "helloworld");

        Thread.sleep(500);

        Thread subThread = new Thread(new Runnable() {
            public void run() {
                try {
                    Jedis jedisOfSub = jedisPool.getResource();
                    SubClient subClient = new SubClient(jedisOfSub);
                    System.out.println("----------subscribe operation begin-------");
                    JedisPubSub listener = new PrintListener();
                    subClient.sub(listener, channel);
                    System.out.println("----------subscribe operation end-------");
                    returnJedis(jedisOfSub);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
        subThread.start();

        int i = 0;
        while (i < 10) {
            String message = String.valueOf(System.currentTimeMillis());
            pubClient.pub(channel, message);
            i++;
            Thread.sleep(1000);
        }
        pubClient.pub(channel, "quit");
        returnJedis(jedisOfPub);
    }

    //生产者消费者模式
    public void procon() {
        try {
            final String key = "testprocon";
            Jedis jedisOfPro = jedisPool.getResource();
            Producer producer = new Producer(jedisPool.getResource());
            for (int i = 0; i < 3; i++) {
                producer.produce("testprocon", (i + 1) + "元");
                System.out.println("发布红包---" + (i + 1) + "元");
            }

            Thread.sleep(100);

            int consumerNums = 10;
            List<Thread> threadList = new ArrayList<Thread>();
            for (int i = 0; i < consumerNums; i++) {
                final int finalI = i;
                Thread consume = new Thread(new Runnable() {
                    public void run() {
                        try {
                            Jedis jedisOfCon = jedisPool.getResource();
                            Consumer consumer = new Consumer(jedisOfCon);
                            System.out.println("消费者" + (finalI + 1) + "号---取值:" + consumer.consume(key));
                            returnJedis(jedisOfCon);
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                });
                threadList.add(consume);
            }
            for (Thread thread : threadList) {
                thread.start();
            }
            returnJedis(jedisOfPro);
        } catch (Exception e) {
            e.printStackTrace();
        }


    }

    public void returnJedis(Jedis jedis) {
        jedisPool.returnBrokenResource(jedis);
    }

}
